package com.overlord.im.commons.mq.config;

import com.overlord.im.commons.mq.RabbitMqConstants;
import org.aopalliance.aop.Advice;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.GenericMessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.retry.policy.MapRetryContextCache;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import javax.annotation.Resource;

/**
 * @author wuwei
 * @className RabbitConfiguration
 * @description
 * @date 2020/4/1 17:23
 **/
@Configuration
public class RabbitConfiguration {
    @Resource
    private RabbitMqProperties rabbitMqProperties;

    @Bean(name = RabbitMqConstants.DEFAULT_MQ_ERROR_HANDLER)
    public MqErrorHandler errorHandler() {
        return new MqErrorHandler();
    }

    @Bean
    public ConnectionFactory rabbitConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(rabbitMqProperties.getHost());
        connectionFactory.setPort(Integer.valueOf(rabbitMqProperties.getPort()).intValue());
        connectionFactory.setUsername(rabbitMqProperties.getUsername());
        connectionFactory.setPassword(rabbitMqProperties.getPassword());
        connectionFactory.setVirtualHost(rabbitMqProperties.getVhost());
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        connectionFactory.setChannelCacheSize(25);
        connectionFactory.setRequestedHeartBeat(500);
        connectionFactory.setConnectionTimeout(30000);
        return connectionFactory;
    }

    @Bean
    public AmqpAdmin rabbitAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitConnectionFactory());
        rabbitAdmin.setAutoStartup(true);
        rabbitAdmin.setIgnoreDeclarationExceptions(false);
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory());
        SimpleMessageConverter simpleMessageConverter = new SimpleMessageConverter();
        simpleMessageConverter.setCreateMessageIds(true);
        rabbitTemplate.setMessageConverter(simpleMessageConverter);
        return rabbitTemplate;
    }

    @Bean
    public MapRetryContextCache mapRetryContextCache() {
        return new MapRetryContextCache();
    }

    @Bean
    public MessageConverter messageConverter() {
        return new GenericMessageConverter();
    }

    @Bean
    public SimpleRabbitListenerContainerFactory multiThreadListenerContainerFactory() {
        return createListenerContainerFactory(
            new ListenerContainerFactoryRegister(AcknowledgeMode.AUTO, Integer.valueOf(10), Integer.valueOf(30),
                Integer.valueOf(10), Integer.valueOf(10)));
    }

    @Bean
    public SimpleRabbitListenerContainerFactory multiThreadTo20ListenerContainerFactory() {
        return createListenerContainerFactory(
            new ListenerContainerFactoryRegister(AcknowledgeMode.AUTO, Integer.valueOf(20), Integer.valueOf(60),
                Integer.valueOf(10), Integer.valueOf(10)));
    }

    @Bean
    public SimpleRabbitListenerContainerFactory multiThreadTo50ListenerContainerFactory() {
        return createListenerContainerFactory(
            new ListenerContainerFactoryRegister(AcknowledgeMode.AUTO, Integer.valueOf(50), Integer.valueOf(150),
                Integer.valueOf(10), Integer.valueOf(10)));
    }

    @Bean
    public SimpleRabbitListenerContainerFactory multiThreadAndManualAcknowledgeListenerContainerFactory() {
        return createListenerContainerFactory(
            new ListenerContainerFactoryRegister(AcknowledgeMode.MANUAL, Integer.valueOf(3), Integer.valueOf(10),
                Integer.valueOf(10), Integer.valueOf(10)));
    }

    @Bean
    public SimpleRabbitListenerContainerFactory singleThreadListenerContainerFactory() {
        return createListenerContainerFactory(
            new ListenerContainerFactoryRegister(AcknowledgeMode.AUTO, Integer.valueOf(1), Integer.valueOf(1),
                Integer.valueOf(1), Integer.valueOf(3)));
    }

    @Bean
    public SimpleRabbitListenerContainerFactory singleThreadAndManualAcknowledgeListenerContainerFactory() {
        return createListenerContainerFactory(
            new ListenerContainerFactoryRegister(AcknowledgeMode.MANUAL, Integer.valueOf(1), Integer.valueOf(1),
                Integer.valueOf(1), Integer.valueOf(3)));
    }

    private SimpleRabbitListenerContainerFactory createListenerContainerFactory(
        ListenerContainerFactoryRegister register) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(rabbitConnectionFactory());
        factory.setAcknowledgeMode(register.getAckMode());
        factory.setConcurrentConsumers(register.getConcurrentConsumers());
        factory.setMaxConcurrentConsumers(register.getMaxConcurrentConsumers());
        factory.setTxSize(register.getTxSize());
        factory.setPrefetchCount(register.getPerFetch());
        factory.setAutoStartup(true);
        factory.setChannelTransacted(true);
        factory.setErrorHandler(errorHandler());
        ContentTypeDelegatingMessageConverter converter = new ContentTypeDelegatingMessageConverter();
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        converter.addDelegate("json", jackson2JsonMessageConverter);
        converter.addDelegate("application/json", jackson2JsonMessageConverter);
        factory.setMessageConverter(converter);
        factory.setAdviceChain(adviceChain(AcknowledgeMode.MANUAL == register.getAckMode()));
        return factory;
    }

    public Advice[] adviceChain(boolean isManual) {
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000);
        backOffPolicy.setMultiplier(2);
        backOffPolicy.setMaxInterval(3000);
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(RabbitMqConstants.RETRY_TIMES));
        StatelessRetryOperationsInterceptorFactoryBean factoryBean =
            new StatelessRetryOperationsInterceptorFactoryBean();
        factoryBean.setRetryOperations(retryTemplate);
        factoryBean.setMessageRecoverer(new RetryFailAckRecover());
        RetryOperationsInterceptor operationsInterceptor = factoryBean.getObject();
        if (isManual) {
            return new Advice[] {new ManualAckInterceptor(), operationsInterceptor};
        }
        return new Advice[] {operationsInterceptor};
    }

    class ListenerContainerFactoryRegister {
        private AcknowledgeMode ackMode;

        private Integer concurrentConsumers;

        private Integer maxConcurrentConsumers;

        private Integer txSize;

        private Integer perFetch;

        public ListenerContainerFactoryRegister(AcknowledgeMode ackMode, Integer concurrentConsumers,
            Integer maxConcurrentConsumers, Integer txSize, Integer perFetch) {
            this.ackMode = ackMode;
            this.concurrentConsumers = concurrentConsumers;
            this.maxConcurrentConsumers = maxConcurrentConsumers;
            this.txSize = txSize;
            this.perFetch = perFetch;
        }

        public AcknowledgeMode getAckMode() {
            return this.ackMode;
        }

        public void setAckMode(AcknowledgeMode ackMode) {
            this.ackMode = ackMode;
        }

        public Integer getConcurrentConsumers() {
            return this.concurrentConsumers;
        }

        public void setConcurrentConsumers(Integer concurrentConsumers) {
            this.concurrentConsumers = concurrentConsumers;
        }

        public Integer getMaxConcurrentConsumers() {
            return this.maxConcurrentConsumers;
        }

        public void setMaxConcurrentConsumers(Integer maxConcurrentConsumers) {
            this.maxConcurrentConsumers = maxConcurrentConsumers;
        }

        public Integer getTxSize() {
            return this.txSize;
        }

        public void setTxSize(Integer txSize) {
            this.txSize = txSize;
        }

        public Integer getPerFetch() {
            return this.perFetch;
        }

        public void setPerFetch(Integer perFetch) {
            this.perFetch = perFetch;
        }
    }
}
